package com.project.pulsar.delayMsgAndSubscriptionsModel;

import com.project.pulsar.conf.PulsarConf;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 共享模式
 */
@RestController
public class ShareMessageController {
    @Autowired
    PulsarConf pulsarConf;

    String tenantName = "delay";//调用接口创建的租户,创建方法见com.project.pulsar.persistent.PersistentController.java
    String namespace = "mySpace";//调用接口创建的命名空间,创建方法见com.project.pulsar.persistent.PersistentController.java

    /**
     * 生产消息
     *
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/share/sendMsg")
    public void sendMsg(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        //持久化         租户   命名空间   主题
        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic(topic)
                .create();
        producer.newMessage().key("key-1").value("message-1-1".getBytes()).send();
        producer.newMessage().key("key-1").value("message-1-2".getBytes()).send();
        producer.newMessage().key("key-1").value("message-1-3".getBytes()).send();
        producer.newMessage().key("key-2").value("message-2-1".getBytes()).send();
        producer.newMessage().key("key-2").value("message-2-2".getBytes()).send();
        producer.newMessage().key("key-2").value("message-2-3".getBytes()).send();
        producer.newMessage().key("key-3").value("message-3-1".getBytes()).send();
        producer.newMessage().key("key-3").value("message-3-2".getBytes()).send();
        producer.newMessage().key("key-4").value("message-4-1".getBytes()).send();
        producer.newMessage().key("key-4").value("message-4-2".getBytes()).send();
    }


    /**
     * 手动执行获取消息
     *
     * @throws PulsarClientException
     */
    @GetMapping("/share/comsumer")
    public void comsumerByArtificial() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        //持久化         租户   命名空间   主题
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)//指定模式
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

    /**
     * 手动执行获取消息
     *
     * @throws PulsarClientException
     */
    @GetMapping("/share/comsumer2")
    public void comsumerByArtificial2() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        //持久化         租户   命名空间   主题
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)//指定模式
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }
}